There appears to be a race condition in the subscription cleanup logic. When checking sub.tx.receiver_count() == 0 and then performing invalidation and UNLISTEN operations, new receivers could be added to the subscription between these steps. This creates a Time-of-Check-Time-of-Use vulnerability where active subscriptions might be incorrectly terminated.

Consider using a more atomic approach for this cleanup operation, such as:

1. Acquiring a lock before checking the receiver count
2. Using a compare-and-swap pattern if available
3. Moving this logic into a synchronized context where subscription additions are also managed

This would prevent the scenario where a subscription is invalidated while new clients are attempting to use it.

Spotted by Diamond



Is this helpful? React 👍 or 👎 to let us know.

The error handling in this connection polling task needs improvement to prevent resource leaks and silent failures. Currently, when deserialization errors occur (lines 96-101 and 108-111), the task breaks out of the loop and terminates without proper cleanup or recovery. This creates several issues:

1. The connection remains in an undefined state when the task terminates early
2. Message processing stops permanently with no notification to subscribers
3. The client connection resource may leak if the task exits unexpectedly

Consider implementing:

• Robust error handling that logs but continues processing other messages
• A reconnection mechanism if the connection fails
• A way to notify the system when message delivery is compromised
• Proper resource cleanup in all termination paths

This would make the system more resilient to transient errors and prevent silent degradation of the pub/sub functionality.

Spotted by Diamond



Is this helpful? React 👍 or 👎 to let us know.